package com.xj.common.sdk.service.kafka;

import cn.hutool.core.convert.Convert;
import com.alibaba.fastjson.JSON;
import com.xj.common.bean.constant.Common;
import com.xj.common.bean.driver.DriverConfiguration;
import com.xj.common.bean.driver.DriverMetadata;
import com.xj.common.model.ControlInfo;
import com.xj.common.sdk.bean.driver.DriverContext;
import com.xj.common.sdk.service.ControlService;
import com.xj.common.sdk.service.DriverMetadataService;
import com.xj.common.sdk.service.DriverService;
import com.xj.common.utils.JsonPaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
* 类说明 :接收KAFKA总线上接收到的控制命令，进行处理
*/
@Slf4j
@Component
public class ControlClientConsumer {
    @Resource
    private DriverContext driverContext;

    @Resource
    private ControlService controlService;
	 // 消费监听
    @KafkaListener(topics = {Common.KAFKA.TOPIC_EXCHANGE_CONTROL_CLIENT},groupId = "ControlClientSDK2")
    public void onStatusMessage(ConsumerRecord<?, ?> record,Acknowledgment ack){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("DriverMetaConsumer简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
        String jsonstr = record.value().toString();
        ControlInfo controlInfo= JsonPaseUtil.JSONStrToObj(jsonstr,  ControlInfo.class);
        //因为kafka没有单独通道，所以必须做设备识别
        if(driverContext.getDriverMetadata().getDeviceSnMap().containsKey(controlInfo.getDeviceSn())){
            controlService.sendCtl(controlInfo);
        }
        //提交偏移量，防止重复消费
        ack.acknowledge();
    }


}
